package study.rocketmq.origin;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.List;

public class Producer {
    public static final String NAMESRC_ADDR = "120.25.73.46:9876";
    public static void main(String[] args) throws Exception {
        //同步发送
        SyncSend();
        //异步发送
        //AsyncSend();
    }

    /**
     * 同步发送消息
     */
    public static void SyncSend()throws Exception{
        DefaultMQProducer producer = new DefaultMQProducer("producerGroup-01");
        producer.setNamesrvAddr(NAMESRC_ADDR);
        /***TODO  队列数量，默认4个 , 不建议在这里设置 ，应该先配置好topic */
        producer.setDefaultTopicQueueNums(4);
        /***TODO  超时时间，默认3秒 */
        producer.setSendMsgTimeout(3000);
        /***TODO  同步发送失败重试次数 默认就是2- 同步发送时  */
        producer.setRetryTimesWhenSendFailed(2);
        /***TODO  异步发送失败重试次数 默认就是2- 异步发送时 */
//        producer.setRetryTimesWhenSendAsyncFailed(2);
        // 客户端回调线程数，默认为CPU核数
        //producer.setClientCallbackExecutorThreads();
        producer.start();
        /**
         * String topic, topic
         * String tags, 标签
         * String keys,  用户自定义的key，一般是消息的唯一标识
         * int flag,
         * byte[] body,
         * boolean waitStoreMsgOK
         */
        String strMessage = "";
        Message message = null;
        String topic = "topic-17";
        for (int i = 0; i < 10; i++) {
            strMessage = "message-"+i;


            if(i > 5){
                message = new Message(topic,"TagB","key-"+i,strMessage.getBytes());
            }else{
                message = new Message(topic,"TagA","key-"+i,strMessage.getBytes());
            }
            /***TODO  延迟消息，只能固定的时间点
             *   messageDelayLevel = 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
             *   ps：消费失败重试从10秒的值开始，后面一样
             * */
            if(i == 2){
                message.setDelayTimeLevel(3);  // 3代表10秒
            }

            SendResult sendResult = producer.send(message,10000);


            /***TODO  消息发送到指定的队列, */
//            SendResult sendResult = producer.send(message, new MessageQueueSelector() {
//
//                @Override
//                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
//                    System.out.println("队列个数-"+mqs.size());
//                    Integer queueNumber = (Integer)arg;
//                    int index = queueNumber % mqs.size();
//                    return mqs.get(index);
//                }
//                //这个2会传递给select方法，队列的下标
//            }, 2);


            /**
             * TODO SendStatus
             *     - SEND_OK, 完全正常，全部成功
             *     - FLUSH_DISK_TIMEOUT, 消息发送成功，但是刷盘超时
             *     - FLUSH_SLAVE_TIMEOUT, 发送到Slave超时
             *     - SLAVE_NOT_AVAILABLE, Slave不可用
             *  高可靠性要求的业务，对于非SEND_OK的状态，需要重新发送。返回其他状态可能产生数据丢失。
             */
            if(sendResult.getSendStatus() == SendStatus.SEND_OK){
                System.out.println("发送成功："+sendResult);
            }else{
                /** TODO 需要重发 */
                System.out.println("发送失败，有丢失数据的风险，一般要重新发送-->");
            }


        }
        System.out.println("-------发送结束------>");
        producer.shutdown();
        System.out.println("-------shutdown结束------>");

    }

    /**
     * 异步发送消息
     */
    public static void AsyncSend()throws Exception{
        DefaultMQProducer producer = new DefaultMQProducer("producerGroup-01");
        producer.setNamesrvAddr(NAMESRC_ADDR);
        /***TODO  队列数量，默认4个 , 不建议在这里设置 ，应该先配置好topic */
        producer.setDefaultTopicQueueNums(4);
        /***TODO  超时时间，默认3秒 */
        producer.setSendMsgTimeout(3000);
        /***TODO  同步发送失败重试次数 默认就是2- 同步发送时  */
        producer.setRetryTimesWhenSendFailed(2);
        /***TODO  异步发送失败重试次数 默认就是2- 异步发送时 */
        producer.setRetryTimesWhenSendAsyncFailed(2);

        producer.start();

        String strMessage = "消息水电费极速对决";
        String topic = "topic-async-01";
        Message message = new Message(topic,"TagA","key-1",strMessage.getBytes());
        //  2.2 异步发送消息
        producer.send(message, new SendCallback() {
            //rabbitmq急速入门的实战: 可靠性消息投递
            @Override
            public void onSuccess(SendResult sendResult) {


                /**
                 * TODO SendStatus
                 *     - SEND_OK, 完全正常，全部成功
                 *     - FLUSH_DISK_TIMEOUT, 消息发送成功，但是刷盘超时
                 *     - FLUSH_SLAVE_TIMEOUT, 发送到Slave超时
                 *     - SLAVE_NOT_AVAILABLE, Slave不可用
                 *  高可靠性要求的业务，对于非SEND_OK的状态，需要重新发送。返回其他状态可能产生数据丢失。
                 */
                if(sendResult.getSendStatus() == SendStatus.SEND_OK){
                    System.out.println("发送成功："+sendResult);
                }else{
                    /** TODO 需要重发 */
                    System.out.println("发送失败，有丢失数据的风险，一般要重新发送-->");
                }
                System.out.println("onSuccess -msgId: " + sendResult.getMsgId() + ", status: " + sendResult.getSendStatus());
            }
            @Override
            public void onException(Throwable e) {
                e.printStackTrace();
                System.err.println("------发送失败");
            }
        });
        System.out.println("-------发送结束------>");
        Thread.sleep(5000);
        producer.shutdown();
        System.out.println("-------shutdown结束------>");

    }

}
